package org.common.utils.consumer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * map消费者，根据map中key唯一消费，不重复队列
 * 创建一个唯实体的，重复id实体，则不会多次添加，消费者组
 * @author zhouzhibing
 * @date 2025/8/1
 */
class ConsumerMapQueue extends ConsumerQueue {

    private static final Logger log = LoggerFactory.getLogger(ConsumerMapQueue.class);

    /**
     * 队列实体<唯一ID，实体>
     */
    private Map<Long , Entry> entityMap = new ConcurrentHashMap<>();

    /**
     * 构造函数
     * @param config 配置
     * @param index 索引
     */
    ConsumerMapQueue(ConsumerConfig config , int index) {
        super(config , index);
    }

    /**
     * 添加异步实体
     * @param entity 异步实体
     */
    public void add(Long key , Object entity) {
        if(!isStart()) {
            throw new ConsumerException("ConsumerQueue.add is close , name = " + getName() + " , entity = " + entity);
        }

        // 如果key已经存在，则不添加
        Entry oldEntry = entityMap.get(key);
        if(oldEntry != null) {
            return;
        }

        // 添加队列
        Entry entry = new Entry(key, entity);
        super.add(entry);
        entityMap.put(key, entry);
    }

    /**
     * 获取队列实体
     * @return 队列实体
     * @throws InterruptedException
     */
    protected Object take() throws InterruptedException {
        Entry entry = null;
        Object entity = null;
        try {
            entry = (Entry)super.take();
            entity = entry.getEntity();
        } finally {
            if(entity != null) {
                entityMap.remove(entry.getKey());
            }
        }
        return entity;
    }

    /**
     * 销毁
     */
    protected void destroy() {
        super.destroy();
        entityMap.clear();
        entityMap = null;
    }

    private static class Entry {
        /**
         * 实体key
         */
        private Long key;
        /**
         * 实体
         */
        private Object entity;

        public Entry(Long key , Object entity) {
            this.key = key;
            this.entity = entity;
        }

        public Long getKey() {
            return key;
        }

        public Object getEntity() {
            return entity;
        }
    }
}